package com.atguigu.api4

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.scala._
import org.apache.flink.table.api.{DataTypes, EnvironmentSettings, Table}
import org.apache.flink.table.descriptors.{Csv, Kafka, Schema}

/**
 * @description: 流式输出
 * @time: 2020/7/22 17:22
 * @author: baojinlong
 **/
object TableApiTest2 {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 默认就是老版本流查询环境
    // val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)

    // 设置并行度方便测试
    env.setParallelism(1)

    // 1.1创建老版本流查询环境
    // 环境参数设置
    val settings: EnvironmentSettings = EnvironmentSettings.newInstance()
      .useOldPlanner()
      .inStreamingMode()
      .build()
    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env, settings)

    // 连接到Kafka
    tableEnv.connect(
      new Kafka()
        .version("0.11")
        .topic("sensor")
        .property("bootstrap.servers", "localhost:9092")
        .property("zookeeper.connect", "localhost:2182")
    )
      .withFormat(new Csv)
      .withSchema(
        new Schema()
          .field("id", DataTypes.STRING)
          .field("timestamp", DataTypes.BIGINT)
          .field("temperature", DataTypes.DOUBLE)
      )
      .createTemporaryTable("kafkaInputTable")

    // 01简单查询,过滤投影
    val sensorTableTmp: Table = tableEnv.from("inputTable")
    // 隐式转换
    sensorTableTmp.select('id, 'temperature)
      .filter('id === "sensor_01")
    // 02sql简单查询
    val sqlResult: Table = tableEnv.sqlQuery(
      """
        |select id,temperature
        |from inputTable
        |where id = 'sensor_01'
        |""".stripMargin)

    // 03简单聚合
    val aggResultTable: Table = sensorTableTmp
      .groupBy('id)
      .select('id, 'id.count as 'count)
    // sql实现简单聚合
    val aggResultSqlTable: Table = tableEnv.sqlQuery("select id,count(id) as cnt from inputTable group by id")
    // 打印输出:注意聚合的结果不能追加打印
    aggResultSqlTable.toRetractStream[(String, Long)].print("aggResult")


    // 转换成流打印输出
    val sensorTable: Table = tableEnv.from("kafkaInputTable")
    sensorTable.toAppendStream[(String, Long, Double)].print("kafkaInputTable-xxx")


    // 执行
    env.execute("table env job")
  }

}
